package cas.ihep.pwa.fcn;

import alluxio.AlluxioURI;
import cas.ihep.alluxio.AlluxioBlock;
import cas.ihep.alluxio.AlluxioFile;
import cas.ihep.alluxio.AlluxioSystem;
import cas.ihep.pwa.evtgen.EvtPDL;
import cas.ihep.spark.JobParser;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

public class BgFcn extends BaseFcn implements Function<String, Tuple2<Double,Double>> ,PairFunction<String,Double,Double>{

    private static final long serialVersionUID=0xcafebabe9264343L;
    public BgFcn(){super();}

//    public BgFcn(Broadcast<EvtPDL> pdl, Broadcast<JobParser> parser,Broadcast<double[]> vpars,Broadcast<String> decay){
//        super(pdl,parser,vpars,decay);
//    }

    public BgFcn(EvtPDL pdl,JobParser jp,double[] v,String decay){
        super(pdl,jp,v,decay);
    }

    @Override
    public Tuple2<Double, Double> call(String tuple) throws Exception {
        mmapDecay(bdecay);
        double ret[]=new double[2];
        AlluxioSystem asys=AlluxioSystem.get();
        AlluxioURI apath=new AlluxioURI(tuple);
        AlluxioFile afile=asys.open(apath);
        System.out.println("Processing Bg file: "+tuple);
        AlluxioBlock ablk=afile.getBlock(0);
        String path=bgcalamps(ablk.getBlockPath(),vpars,ret,pdl,parser,parser);
        ablk.close();
        afile.close();
        asys.getFileSystem().delete(apath);
        afile=asys.create(apath);
        ablk=afile.nextBlock();
        link(path,ablk.getBlockPath());
        ablk.close();
        afile.close();
        return new Tuple2<>(ret[0],ret[1]);
    }
}
